package njuflink2;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;
import org.apache.kafka.common.serialization.StringSerializer;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

public class CopySplit {
    public static void main(String[] args) throws Exception {
        //获取flink运行环境
        StreamExecutionEnvironment env=StreamExecutionEnvironment.getExecutionEnvironment();
        //指定topic名称
        String topic = njuflink2.GlobalConfig.TOPIC2;
        //Kafka配置参数
        Properties properties=new Properties();
        properties.setProperty("bootstrap.servers", njuflink.GlobalConfig.BOOTSRAP_SEVERS);
        properties.setProperty("group.id",topic);
        //指定key value的序列化方式
        properties.put("key.serializer", StringSerializer.class.getName());
        properties.put("value.serializer", StringSerializer.class.getName());

        //Kafka消费者
        FlinkKafkaConsumer<String> myConsumer=new FlinkKafkaConsumer<String>(topic,new SimpleStringSchema(),properties);
        DataStream<String> stream=env.addSource(myConsumer);
        System.out.println("connect kafka");
        System.out.println(myConsumer);


        DataStream<Tuple2<String,Long>>langnewsCounts1=stream.flatMap((String line, Collector<Tuple2<String,Long> > out)->{
            String []get2=line.split(",");
            String tmplang=get2[0];
            if(!tmplang.equals("Unknown Language")){
                out.collect(Tuple2.of(tmplang,1L));
            }

        })
                .returns(Types.TUPLE(Types.STRING,Types.LONG));
        DataStream<Tuple2<String,Long>>licnewsCounts1=stream.flatMap((String line, Collector<Tuple2<String,Long> > out)->{
            String []get2=line.split(",");
            String tmplic=get2[1];
            if(!tmplic.equals("Unknown License")){
                out.collect(Tuple2.of(tmplic,1L));
            }

        })
                .returns(Types.TUPLE(Types.STRING,Types.LONG));
        System.out.println("map");
        init_tb1();
        init_tb2();


        KeyedStream<Tuple2<String,Long>,String> langnewCountsStream=langnewsCounts1.keyBy(data->data.f0);
        KeyedStream<Tuple2<String, Long>,String > licnewCountsStream=licnewsCounts1.keyBy(data->data.f0);
        SingleOutputStreamOperator<Tuple2<String, Long>> languageSum = langnewCountsStream.sum(1);
        SingleOutputStreamOperator<Tuple2<String, Long>> licenseSum = licnewCountsStream.sum(1);
        System.out.println("languageSum:");
        System.out.println(languageSum);
        languageSum.print();
        System.out.println("licenseSum:");
        System.out.println(licenseSum);
        licenseSum.print();
//        languageSum.addSink(new MySQLSink2_1());
//        licenseSum.addSink(new MySQLSink2_2());

//        //自定义MySQLSink
//        periodCounts.addSink(new MySQLSink2());

        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
                3, // 尝试重启的次数
                Time.of(10, TimeUnit.SECONDS) // 间隔
        ));
        //执行Flink程序
        env.execute();
    }
    public static void init_tb1() throws ClassNotFoundException, SQLException {
        Connection connection;
        PreparedStatement preparedStatement;
        String tb= njuflink2.GlobalConfig.TABLE2_1;
        //        //加载JDBC驱动
        Class.forName(njuflink.GlobalConfig.DRIVER_CLASS);
        //获取数据库链接
        connection= DriverManager.getConnection(njuflink.GlobalConfig.DB_URL, njuflink.GlobalConfig.USER_NAME, njuflink.GlobalConfig.PASSWORD);
        // 关闭自定提交
        connection.setAutoCommit(false);
        String createsql="CREATE TABLE IF NOT EXISTS "+tb+"(language VARCHAR(50), count INT(11));";
        System.out.println(createsql);
        preparedStatement=connection.prepareStatement(createsql);
        preparedStatement.executeUpdate();
//        preparedStatement=connection.prepareStatement("SHOW TABLES LIKE \""+tb+"\"");
//        ResultSet rs=preparedStatement.executeQuery();
//        if(rs!=null&&rs.getFetchSize()!=0){
//            System.out.println(tb+"表已存在");
//            String droptbsql = "DROP TABLE "+tb;
//            preparedStatement.executeUpdate(droptbsql);
//            System.out.println(tb+"表已删除");
//        }
//        preparedStatement=connection.prepareStatement("CREATE TABLE "+tb+"(language VARCHAR(25), count INT(11)) " );
//        preparedStatement.executeUpdate();
        System.out.println(tb+"表创建成功");
        connection.commit();
        if(preparedStatement!=null){
            preparedStatement.close();
        }
        if(connection!=null){
            connection.close();
        }
    }
    public static void init_tb2() throws ClassNotFoundException, SQLException {
        Connection connection;
        PreparedStatement preparedStatement;
        String tb= njuflink2.GlobalConfig.TABLE2_2;
        //        //加载JDBC驱动
        Class.forName(njuflink.GlobalConfig.DRIVER_CLASS);
        //获取数据库链接
        connection= DriverManager.getConnection(njuflink.GlobalConfig.DB_URL, njuflink.GlobalConfig.USER_NAME, njuflink.GlobalConfig.PASSWORD);
        // 关闭自定提交
        connection.setAutoCommit(false);
        String createsql="CREATE TABLE IF NOT EXISTS "+tb+"(license VARCHAR(50), count INT(11));";
        System.out.println(createsql);
        preparedStatement=connection.prepareStatement(createsql);
        preparedStatement.executeUpdate();
//        preparedStatement=connection.prepareStatement("SHOW TABLES LIKE \""+tb+"\"");
//        ResultSet rs=preparedStatement.executeQuery();
//        if(rs!=null&&rs.getFetchSize()!=0){
//            System.out.println(tb+"表已存在");
//            System.out.println(rs.getFetchSize());
//            String droptbsql = "DROP TABLE "+tb;
//            preparedStatement.executeUpdate(droptbsql);
//            System.out.println(tb+"表已删除");
//        }
//        preparedStatement=connection.prepareStatement("CREATE TABLE "+tb+"(license VARCHAR(25), count INT(11)) " );
//        preparedStatement.executeUpdate();
        System.out.println(tb+"表创建成功");
        connection.commit();
        if(preparedStatement!=null){
            preparedStatement.close();
        }
        if(connection!=null){
            connection.close();
        }
    }
}
